﻿using Leo.Common.MQ;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Hosting;

namespace Leo.Common.Kafka
{
    public class KafkaConsumerHostedService : BackgroundService
    {
        public KafkaConsumerHostedService(KafkaBuilder kafkaBuilder, IServiceProvider services)
        {
            this.kafkaBuilder = kafkaBuilder;
            this.services = services;
        }


        protected override Task ExecuteAsync(CancellationToken stoppingToken)
        {
            foreach (KafkaConsumerWrapper kafkaConsumerWrapper in this.kafkaBuilder.wrappers)
            {
                IMessageHandler[] handles = (from x in kafkaConsumerWrapper.HandlerTypes
                                             select (IMessageHandler)ServiceProviderServiceExtensions.GetRequiredService(this.services, x)).ToArray<IMessageHandler>();
                kafkaConsumerWrapper.Config.Broker = this.kafkaBuilder.Setting.Broker;
                KafkaConsumer kafkaConsumer = new KafkaConsumer(kafkaConsumerWrapper.Config, stoppingToken, handles);
                this.kafkaBuilder.Consumers.Add(kafkaConsumer);
                kafkaConsumer.Start();
            }
            return Task.CompletedTask;
        }


        private readonly IServiceProvider services;


        private readonly KafkaBuilder kafkaBuilder;
    }
}
